package com.eco.fanliapp.shengyaappeurekeclient.rocketmq;


import com.alibaba.fastjson.JSON;
import com.aliyun.mq.http.MQClient;
import com.aliyun.mq.http.MQConsumer;
import com.aliyun.mq.http.common.AckMessageException;
import com.aliyun.mq.http.model.Message;
import com.eco.fanliapp.rocketmq.RocketConfig;
import com.eco.fanliapp.shengyaappeurekeclient.controller.LuceneController;
import com.eco.fanliapp.shengyaappeurekeclient.utils.LuceneUtil;
import lombok.Synchronized;
import lombok.extern.slf4j.Slf4j;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.index.IndexWriterConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Component;
import org.wltea.analyzer.lucene.IKAnalyzer;

import javax.annotation.PostConstruct;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * rocketmq生产者启动初始化类
 */
@Component
@Order(value = 2)
@Slf4j
public class RocketmqConsumerInit implements CommandLineRunner {

    @Autowired
    private LuceneController luceneController;
    @Autowired
    private LuceneUtil luceneUtil;
    @Autowired
    private RedisTemplate redisTemplate;

    private static MQConsumer consumer;

    @PostConstruct
    public void init() {
        MQClient mqClient = new MQClient(RocketConfig.NAMESRV_ADDR, RocketConfig.AccessKey, RocketConfig.SecretKey);
        if (RocketConfig.INSTANCEID != null && RocketConfig.INSTANCEID != "") {
            consumer = mqClient.getConsumer(RocketConfig.INSTANCEID, RocketConfig.TOPIC, RocketConfig.GROUP_ID, RocketConfig.TAG);
        } else {
            consumer = mqClient.getConsumer(RocketConfig.TOPIC, RocketConfig.GROUP_ID);
        }
        log.info("consumer is start !!! GROUP_ID:{},TOPIC:{},NAMESRV_ADDR:{}", RocketConfig.GROUP_ID, RocketConfig.TOPIC, RocketConfig.NAMESRV_ADDR);
    }

    /**
     * 初始化消费者
     *
     * @return
     */
    public MQConsumer getconsumer() {
        return this.consumer;
    }

    @Override
    public void run(String... args) throws Exception {
        redisTemplate.setKeySerializer(RedisSerializer.string());
        redisTemplate.setHashKeySerializer(RedisSerializer.string());
        redisTemplate.setHashValueSerializer(RedisSerializer.json());
        redisTemplate.setValueSerializer(RedisSerializer.json());
//        Analyzer analyzer = new IKAnalyzer();//中文分词
//        //创建索引写入配置
//        IndexWriterConfig indexWriterConfig = new IndexWriterConfig(analyzer);
//        //创建索引写入对象
//        IndexWriter indexWriter = new IndexWriter(luceneUtil.getDirectory(), indexWriterConfig);
        String dir = null;
        String key = "luceneIndex";
        String pathKey = "lucenePath";
        do {
            if (redisTemplate.hasKey(pathKey)) {
                Map<String, String> map = (Map) redisTemplate.opsForValue().get(pathKey);
                dir = map.get("oldPath");
            }
            IndexWriter indexWriter = luceneUtil.getIndexWriter(dir);
            if (redisTemplate.hasKey(key)) {
                Map<String, Boolean> status = (Map) redisTemplate.opsForValue().get(key);
                for (String itemkey : status.keySet()) {
//                    if (itemkey.equals(dateNowStr)) {
                    if (status.get(itemkey)) {
                        List<Message> messages = null;
                        try {
                            // 长轮询消费消息
                            // 长轮询表示如果topic没有消息则请求会在服务端挂住3s，3s内如果有消息可以消费则立即返回
                            messages = getconsumer().consumeMessage(
                                    1,// 一次最多消费3条(最多可设置为16条)
                                    30
                            );
                        } catch (Throwable e) {
                            e.printStackTrace();
                        }

                        // 没有消息
                        if (messages == null || messages.isEmpty()) {
                            continue;
                        }
                        try {
                            // 处理业务逻辑
                            for (Message message : messages) {
                                String messageBoby = message.getMessageBodyString();
                                String type = JSON.parseObject(messageBoby).getString("type");
                                String itemids = null;
                                String appGoods = null;
                                if (type.equals("delete")) {
                                    itemids = JSON.parseObject(messageBoby).getString("itemids");
                                } else {
                                    appGoods = JSON.parseObject(messageBoby).getString("appGoods");
                                }
                                if (type.contains("update")) {
//                        int i = luceneController.deleteIndex(appGoods, indexWriter);
//                        if (i == 1 || i == 2) {
//                            luceneController.addIndex(appGoods, indexWriter);
//                        }
                                    int i = luceneController.updateIndex(appGoods, indexWriter);
                                } else if (type.contains("add")) {
                                    luceneController.addIndex(appGoods, indexWriter, dir);
                                } else if (type.contains("delete")) {
                                    luceneController.deleteIndexs(itemids, indexWriter, dir);
                                }
                            }
                        } catch (Throwable e) {
                            indexWriter.commit();
                            indexWriter.close();
                            e.printStackTrace();
                            continue;

                        }
                        // Message.nextConsumeTime前若不确认消息消费成功，则消息会重复消费
                        // 消息句柄有时间戳，同一条消息每次消费拿到的都不一样
                        {
                            List<String> handles = new ArrayList<String>();
                            for (Message message : messages) {
                                handles.add(message.getReceiptHandle());
                            }
                            try {
                                getconsumer().ackMessage(handles);
                            } catch (Throwable e) {
                                // 某些消息的句柄可能超时了会导致确认不成功
                                if (e instanceof AckMessageException) {
                                    AckMessageException errors = (AckMessageException) e;
                                    log.info("Ack message fail, requestId is:" + errors.getRequestId() + ", fail handles:");
                                    if (errors.getErrorMessages() != null) {
                                        for (String errorHandle : errors.getErrorMessages().keySet()) {
                                            log.info("Handle:" + errorHandle + ", ErrorCode:" + errors.getErrorMessages().get(errorHandle).getErrorCode()
                                                    + ", ErrorMsg:" + errors.getErrorMessages().get(errorHandle).getErrorMessage());
                                        }
                                    }
                                    continue;
                                }
                                e.printStackTrace();
                            }
                        }
                        try {
                            Thread.sleep(2000);
                        } catch (InterruptedException e1) {
                            e1.printStackTrace();
                        }
                    }
                }
            }

        }
        while (true);

    }

}
